package com.salmon.study.actors;


import akka.actor.UntypedAbstractActor;
import com.salmon.study.message.ReduceData;
import com.salmon.study.message.Result;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public class AggregateActor extends UntypedAbstractActor {

    private Map<String, Integer> finalReduceMap = new HashMap<>();

    @Override
    public void onReceive(Object message) throws Throwable {
        log.info("{}", message);
        if(message instanceof ReduceData) {
            ReduceData reduceData = (ReduceData) message;
            aggregateInMemoryReduce(reduceData.getReduceMap());
        } else if(message instanceof Result) {
            System.out.println(finalReduceMap.toString());
        } else {
            unhandled(message);
        }
    }

    private void aggregateInMemoryReduce(HashMap<String, Integer> reduceMap) {
        Integer count;
        for(String key : reduceMap.keySet()) {
            if(finalReduceMap.containsKey(key)) {
                count = reduceMap.get(key) + finalReduceMap.get(key);
                finalReduceMap.put(key, count);
            } else {
                finalReduceMap.put(key, reduceMap.get(key));
            }
        }
    }
}
